package org.niit.rdd

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
/*
 RDD的算子分为两种
    行动算子 (方法、函数)

    转换算子(方法、函数)
 */
object ActionRDD02 {

  def main(args: Array[String]): Unit = {
    //准备环境
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("Spark")
    val sparkContext = new SparkContext(sparkConf)
    sparkContext.setLogLevel("ERROR")

    //行动算子
     /*
      1.reduce:聚合RDD中的所有元素。先聚合分区内元素，再聚合分区间元素
      */
     val rdd1: RDD[Int] = sparkContext.makeRDD(List(1, 2, 3, 4), 2) // 【1，2】=1+2    【3，4】=3+4   3+7 = 10
      //rdd1.reduce((x:Int,y:Int) => x+ y)
      val i: Int = rdd1.reduce(_ + _)
      println(i)

    /*
    2.collect:在driver中，将RDD转换为数组
     */
    val arrs1: Array[Int] = rdd1.collect()
    println(arrs1.mkString(","))

    /*
    3.count:获取RDD元素的个数
     */
    val count: Long = rdd1.count()
    println(count)

    /*
    4.first:返回RDD中的第一个元素
     */
    val first: Int = rdd1.first()
    println(first)

    /*
    5.take:返回一个由RDD的前N个元素组成数组
     */
    val arrs2: Array[Int] = rdd1.take(3)
    println(arrs2.mkString(","))

    /*
    6.takeOrdered: 对该RDD元素排序后返回前N个元素
     */
    val rdd2: RDD[Int] = sparkContext.makeRDD(List(4, 1, 3, 5, 6, 7, 2))
    val ints: Array[Int] = rdd2.takeOrdered(4)
    println(ints.mkString(","))

    /*
    7.aggregate: 分区的数据通过初始值和分区内的数据进行聚合，然后再和初始值进行分区间的聚合
     */
    val i1: Int = rdd1.aggregate(10)(_ + _, _ + _)//【1，2】=1+2+10=13    【3，4】=3+4+10=17     17+13+10 = 40
    println(i1)
    val i2: Int = rdd1.fold(10)(_ + _) // 1+2+10+3+4 = 20  1+2+10+3+4+10 = 30   1+2+10+3+4+10+10 = 40
    println(i2)//40

    /*
    8.countByKey:统计每种Key出现的次数
     */
    val rdd3: RDD[(String, Int)] = sparkContext.makeRDD(List(("a", 1), ("a", 2), ("b", 3)))
    val resRdd: collection.Map[String, Long] = rdd3.countByKey()
    println(resRdd)


    /*
    save相关算子
        saveAsTextFile rdd可以是任意类型
        saveAsObjectFile  rdd可以是任意类型  将RDD保存为对象 文件中的数据是乱码
        saveAsSequenceFile rdd必须是KV类型  将RDD保存为序列内容 文件中的数据是乱码
     */
    rdd3.saveAsTextFile("output1")
    rdd3.saveAsObjectFile("output2")
    rdd3.saveAsSequenceFile("output3")

  }


}
